package com.atguigu.gmall.realtime.app

import java.util

import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall.realtime.utils.{MyKafkaUtils, MyOffsetUtils, MyRedisUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.{Jedis, JedisPool}

/**
 * 业务数据消费分流任务
 * 1. 准备实时环境
 * 2. 从Redis中读取offset
 * 3. 从Kafka中消费数据
 * 4. 提取offset
 * 5. 处理数据
 *    5.1 转换结构
 *    5.2 分流
 * 6. 刷写kafka缓冲区
 * 7. 提交offset
 */
object DbBaseApp {
  def main(args: Array[String]): Unit = {
    /*准备实时环境*/
    val dbSparkConf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("dbSpark")
    val ssc: StreamingContext = new StreamingContext(dbSparkConf, Seconds(5))

    /*
    *从Redis中读取offset
     */
    val topic: String = "ODS_BASE_DB_0212"
    val groupId: String = "ODS_BASE_DB_GROUP"
    val offsets: Map[TopicPartition, Long] = MyOffsetUtils.ReadOffset(topic, groupId)
    /*
    从kafka中取消费数据
     */
    var kafkaDstream: InputDStream[ConsumerRecord[String, String]] = null
    if (offsets.nonEmpty) {
      kafkaDstream = MyKafkaUtils.GetConsumerDStream(ssc, topic, groupId, offsets)
    } else {
      kafkaDstream = MyKafkaUtils.GetConsumerDStream(ssc, topic, groupId)
    }
    /*
    *提取offset
     */
    var ranges: Array[OffsetRange] = null
    val currentKafkaDstream: DStream[ConsumerRecord[String, String]] = kafkaDstream.transform(
      rdd => {
        ranges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )
    /*
    *提取kafka的value值
     */
    val valueDstream: DStream[JSONObject] = currentKafkaDstream.map(
      data => {
        val jsonString: String = data.value()
        val json: JSONObject = JSON.parseObject(jsonString)
        json
      }
    )
    //验证数据打通
//    println("---------")
 //   valueDstream.print(100)

    /*
      处理数据
    */
    //jsonObjDStream.print(100)
    //5.2 分流
    // 分流原则:
    // 事实表数据:   order_info => DWD_ORDER_INFO_0212_I
    //                         => DWD_ORDER_INFO_0212_U
    //                         => DWD_ORDER_INFO_0212_D
    // 维度表数据:  分流到Redis中
    //TODO 如何动态维护表清单?
    /*
    将表清单的内容存储到redis中实现动态存储
    */

    // 将表清单维护到redis中,代码中只要保证能够周期性的读取redis中维护的表清单.
    // Redis中如何存储:
    // type:     set
    // key :     FACT:TABLES   DIM:TABLES
    // value:   表名的集合
    // 写入API:  sadd
    // 读取API:  smembers
    // 是否过期: 不过期
    //该数据提前在redis数据库中建好

    //事实表清单
    /* val factTables: Set[String] = Set[String]("order_info" , "order_detail" 需要什么表写什么表)
     //维度表清单
     val dimTables: Set[String] = Set[String]("user_info" , "base_province" /*需要什么表写什么表*/)*/
    valueDstream.foreachRDD(
      rdd => {
        /*
          获取在redis中存储的表的数据
          将表名数据设置成广播变量，防止当表名数据过大是，excutor中存在多个重复变量
          一个excutor可能存在多个分区每个分区也就是多个task,每个task都会拉取一份变量，造成重复
         */
        val tableJredis: Jedis = MyRedisUtils.get()
        val factTableNames: util.Set[String] = tableJredis.smembers("FACT:TABLES")
        val dimTableNames: util.Set[String] = tableJredis.smembers("DIM:TABLES")

        val factTableNamesBroad: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(factTableNames)
        val dimTableNamesBroad: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(dimTableNames)

        rdd.foreachPartition(
          dataitor => {
            val jedis = MyRedisUtils.get()
            for (jsonObject <- dataitor) {
              /*
          *获取数据类型
           */
              val typeData: String = jsonObject.getString("type")
              var op = typeData match {
                case "bootstrap-insert" => "I"
                case "insert" => "I"
                case "update" => "U"
                case "delete" => "D"
                case _ => null
              }

              if (op != null) {
                //提取表名
                val tableName: String = jsonObject.getString("table")
                //提取data
                val dataObj: JSONObject = jsonObject.getJSONObject("data")
                //判断是事实表还是维度表
                val factTables: util.Set[String] = factTableNamesBroad.value
                if (factTables.contains(tableName)) {
                  //事实表数据
                  //发往对应的主题
                  val topicName: String = s"DWD_${tableName.toUpperCase()}_0212_$op"
                  MyKafkaUtils.send(topicName, dataObj.toJSONString)
                }
                val dimTables: util.Set[String] = dimTableNamesBroad.value
                if (dimTables.contains(tableName)) {
                  //TODO 历史维度数据如果不发生修改， 怎么保证进入到dim层(redis中)
                  /*
                  *在实时任务进行前，进行一次maxwell的全量同步即可，处理一下"bootstrap-insert"类型的数据
                   */
                  //维度表数据
                  //Redis中如何存储
                  //type:  string
                  //key :  DIM:[tableName]:[id]
                  //value: json
                  //写入API: set
                  //读取API: get
                  //是否过期: 不过期

                  //从dataObj中获取id
                  val dataId: String = dataObj.getString("id")
                  // TODO  此处获取Redis连接好不好?
                  /*
                  *不好，会导致频繁生成连接，应该放在循环后面
                   */
                  /*  val jedis = MyRedisUtils.get()
                      。。。
                     MyRedisUtils.close(jedis)*/
                  val redisKey: String = s"DIM:${tableName.toUpperCase()}:$dataId"
                  jedis.set(redisKey, dataObj.toJSONString)
                }
              }
            }
            MyRedisUtils.close(jedis)
            // *此处进行kafka刷新操作
            MyKafkaUtils.flush()
          }
        )
        //*此处提交offset
        MyOffsetUtils.SaveOffset(topic, groupId, ranges)
      }
    )
    ssc.start()
    ssc.awaitTermination()
  }
}
